/*
 * MIT License
 *
 * Copyright (c) 2020 wen.gu <454727014@qq.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

 /***************************************************************************
 * Name: ipc_service_impl_msvc.cpp
 *
 * Purpose: icp service end implementation on windows platform
 *
 * Developer:
 *   wen.gu , 2020-5-18
 *
 * TODO:
 *
 ***************************************************************************/

 /******************************************************************************
 **    INCLUDES
 ******************************************************************************/
#if defined(_MSC_VER)
#include "panda/sys/ipc_service.h"

#include <thread>
#include <map>

#include "ipc_socket_utils_msvc.h"

#pragma comment(lib,"wsock32.lib")

#include "panda/core/utils.h"

#define LOG_TAG "ipcc"
#include "panda//core/log.h"
 
namespace panda
{
namespace sys
{
/******************************************************************************
 **    MACROS
 ******************************************************************************/
#define SOCK_PATH_PREFIX "/tmp/"
#define MAX_CLIENT_CONNECT_LIMIT 128

/** is the size of socket receive cache buffer */
#define SOCK_RECV_SIZE 1500

/** how many second will be block when call select */
#define SELECT_TIME_OUT_SEC 1
/******************************************************************************
 **    VARIABLE DEFINITIONS
 ******************************************************************************/
using socket_t = int32_t;

struct IpcClientInfo
{
    int32_t id;  /** client id */
    socket_t mScoket;
    std::string name; /** the name of client */
};

using IpcClientInfoMap = std::map<int32_t, IpcClientInfo>;


class IpcService::impl
{
public:
    bool mIsStarted = false;
    bool mIsRunning = true;
    socket_t mSocket = INVALID_SOCKET;
    DataReceiveHandler mReceive;
    std::thread* mThd = nullptr;
    std::string mSrvName;
    ConnectionStateHandler mConnectionStateHandler;
    
    IpcClientInfoMap mClientInfos;

public:
    PandaErrc createSocket(const std::string& srvAddr);

    PandaErrc startReceive(DataReceiveHandler cb);
    void stopRecive();

    PandaErrc sendData(int32_t clientId, const uint8_t* data, uint32_t size);

protected:
    void onReceive();


private:
    void onConnectStateChange(int32_t clientId, bool connected);

    void processConnectRequest(fd_set& sets);
    void processClientRequest(fd_set& sets);

    void addClient(int32_t sockFd, const char* addr);
    void delClient(int32_t clientId);

    socket_t processSocketFd(fd_set& set);

    PandaErrc send_l(int32_t clientId, const uint8_t* data, uint32_t size);
};

/******************************************************************************
 **    inner FUNCTION DEFINITIONS
 ******************************************************************************/

IpcService::PandaErrc IpcService::impl::createSocket(const std::string& srvAddr)
{
    SocketInitializer::get();

    if (mSocket != INVALID_SOCKET)
    {
        LOGE("serivce socket already exist, please destroy old one before create operation\n");
        return PandaErrc::Undefined;
    }
    if ((mSocket = socket(AF_UNIX, SOCK_STREAM, 0)) < 0)
    {
        LOGE("create AF_UNIX socket failed\n");
        mSocket = INVALID_SOCKET;
        return PandaErrc::Undefined;
    }

    struct sockaddr_un srv_addr;
    memset(&srv_addr, 0, sizeof(srv_addr));            /* fill socket address structure with our address */
    srv_addr.sun_family = AF_UNIX;

    snprintf(srv_addr.sun_path, sizeof(struct sockaddr_un), SOCK_PATH_PREFIX "%s", srvAddr.c_str());

    int32_t len = offsetof(struct sockaddr_un, sun_path) + strlen(srv_addr.sun_path) + 1;
    printf("client addr:%s, len:%d\n", srv_addr.sun_path, len);

    int32_t ret = bind(mSocket, (struct sockaddr*) & srv_addr, len);
    if (ret < 0)
    {
        LOGE("bind service socket:%s, failed: %ld\n", srv_addr.sun_path, WSAGetLastError());
        closesocket(mSocket);
        mSocket = INVALID_SOCKET;
        return PandaErrc::Undefined;
    }

    ret = listen(mSocket, MAX_CLIENT_CONNECT_LIMIT);

    if (ret < 0)
    {
        LOGE("listen socket:%s, failed:%ld\n", srv_addr.sun_path, WSAGetLastError());
        closesocket(mSocket);
        mSocket = INVALID_SOCKET;
        return PandaErrc::Undefined;
    }

    return PandaErrc::OK;
}

IpcService::PandaErrc IpcService::impl::startReceive(DataReceiveHandler cb)
{
    mIsRunning = true;
    mReceive = cb;
    mThd = new std::thread(&IpcService::impl::onReceive, this);
    mIsStarted = true;
    return PandaErrc::OK;
}

void IpcService::impl::stopRecive()
{
    if (mIsStarted)
    {
        mIsStarted = false;
        if (mThd)
        {
            mIsRunning = false;
            mThd->join();
        }

        for (auto it : mClientInfos)
        {
            IpcClientInfo& ici = it.second;
            closesocket(ici.mScoket);
        }

        mClientInfos.clear();

        if (mSocket != INVALID_SOCKET)
        {
            closesocket(mSocket);
            mSocket = INVALID_SOCKET;
        }
    }
}

void IpcService::impl::onReceive()
{
    uint8_t recvBuf[2048];

    struct timeval timeout;  //timeout time    
    timeout.tv_sec = SELECT_TIME_OUT_SEC;   /** */
    timeout.tv_usec = 0;
    while (mIsRunning)
    {  
        fd_set server_fd_set;
        socket_t maxFd = processSocketFd(server_fd_set);
        
        int32_t ret = select(maxFd + 1, &server_fd_set, NULL, NULL, &timeout);

        if (ret > 0)
        {
            processConnectRequest(server_fd_set);
            processClientRequest(server_fd_set);
        }
        else if (ret == 0)
        {
            /** timeout, todo something */
        }
        else /** ret < 0 */
        {
            LOGE("srv:%s, do select failed:%ld\n", mSrvName.c_str(), WSAGetLastError());
            /** todo something */
        }
    }
}

void IpcService::impl::onConnectStateChange(int32_t clientId, bool connected)
{
    if (mConnectionStateHandler)
    {
        mConnectionStateHandler(clientId, connected);
    }
}

void IpcService::impl::processConnectRequest(fd_set& sets)
{
    if (FD_ISSET(mSocket, &sets))
    {
        struct sockaddr_un client_addr;
        memset(&client_addr, 0, sizeof(client_addr));
        int32_t len = sizeof(client_addr);
        int32_t client_sock_fd = accept(mSocket, (struct sockaddr*) & client_addr, &len);

        if (client_sock_fd > 0)
        {
            addClient(client_sock_fd, client_addr.sun_path);
            /** todo something */
        }
        else
        {
            LOGE("srv:%s, accept client info failed:%ld\n", mSrvName.c_str(), WSAGetLastError());
        }
    }
}

void IpcService::impl::processClientRequest(fd_set& sets)
{ 
    uint8_t recvBuf[2048]; /**todo refine me, e.g. shared_ptr?? */
    for (auto it : mClientInfos)
    {
        IpcClientInfo& ici = it.second;
        if (FD_ISSET(ici.mScoket, &sets))
        {
            int32_t retSize = ::recv(ici.mScoket, (char*)recvBuf, SOCK_RECV_SIZE, 0);
            if (retSize > 0)
            {
                if (mReceive)
                {
                    mReceive(ici.id, recvBuf, retSize);
                }
            }
            else if (retSize == 0)
            {/** target client is disconnected from current service */
                delClient(ici.id);
            }
            else
            {/** error occur */
                if (errno != EINTR)
                {
                    delClient(ici.id);
                }
                /** todo something */
            }

        }
    }
}


void IpcService::impl::addClient(int32_t sockFd, const char* addr)
{
    IpcClientInfo ici;
    ici.id = sockFd;
    ici.mScoket = sockFd;
    ici.name = addr;
    mClientInfos[sockFd] = ici;
    onConnectStateChange(sockFd, true);
}

void IpcService::impl::delClient(int32_t clientId)
{
    auto it = mClientInfos.find(clientId);

    if (it != mClientInfos.end())
    {
        onConnectStateChange(it->second.id, false);
        closesocket(it->second.mScoket);
        mClientInfos.erase(it);
    }
}

socket_t IpcService::impl::processSocketFd(fd_set& set)
{
    socket_t maxFd = mSocket;
    FD_ZERO(&set);
    FD_SET(mSocket, &set);


    for (auto it : mClientInfos)
    {
        IpcClientInfo& ici = it.second;
        FD_SET(ici.mScoket, &set);
        if (ici.mScoket > maxFd)
        {
            maxFd = ici.mScoket;
        }
    }

    return maxFd;
}


IpcService::PandaErrc IpcService::impl::sendData(int32_t clientId, const uint8_t* data, uint32_t size)
{
    if (clientId >= 0)
    {
        PandaErrc ret = send_l(clientId, data, size);
        if (ret != panda::core::PandaErrc::OK)
        {/** client maybe disconnected */
            delClient(clientId);
        }

        return ret;
    }
    else
    { /** for broadcast mode */
        for (auto it : mClientInfos)
        {
            IpcClientInfo& ici = it.second;
            PandaErrc ret = send_l(ici.mScoket, data, size);
            if (ret != panda::core::PandaErrc::OK)
            {/** client maybe disconnected */
                delClient(clientId);
            }
        }
    }

    return PandaErrc::OK; /**todo, refineme, now aways return OK in broadcast mode */
}   

IpcService::PandaErrc IpcService::impl::send_l(int32_t sockFd, const uint8_t* data, uint32_t size)
{
    while (size > 0)
    {
        int32_t sendSize = (size >= SOCK_RECV_SIZE) ? SOCK_RECV_SIZE : size;   /** is max socket send cache buffer size?? */

        int32_t retSize = ::send(sockFd, (char*)data, sendSize, 0);
        if (retSize > 0)
        {
            size -= retSize;
            data += retSize;
        }
        else if (retSize == 0)
        {
            return PandaErrc::Undefined;
        }
        else
        {
            if (errno != EINTR)
            {
                return PandaErrc::Undefined;    
            }
            /** todo something */
        }

    }

    return PandaErrc::OK;
}

/******************************************************************************
 **    FUNCTION DEFINITIONS
 ******************************************************************************/
IpcService::IpcService(const std::string& srvName, ConnectionStateHandler csh)
    :mImpl(new impl)
{
    /**todosomething */
    mImpl->mSrvName = srvName;
    mImpl->mConnectionStateHandler = csh;
}

IpcService::~IpcService()
{
    stop();
    /**todo something */
}

IpcService::PandaErrc IpcService::start(DataReceiveHandler cb)
{
    if (mImpl->mIsStarted == false)
    {
        PandaErrc ret = mImpl->createSocket(mImpl->mSrvName);

        if (bool(ret))
        {
            ret = mImpl->startReceive(cb);
        }

        return ret;
    }

    return PandaErrc::InvalidStatus;
}

void IpcService::stop()
{
    mImpl->stopRecive();
}

/**
    *  clientId: -1,like broadcast,send to all clients.
    */
IpcService::PandaErrc IpcService::sendData(int32_t clientId, const uint8_t* data, uint32_t size)
{
    return mImpl->sendData(clientId, data, size);
}


} /** namespace sys */
} /** namespace panda */
#endif /** defined(_MSC_VER) */